package com.opages.transaction.message.daemon.scheduled;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;

import com.opages.transaction.message.api.TransactionMessageApi;
import com.opages.transaction.message.daemon.config.MessageProperties;
import com.opages.transaction.message.enums.MessageStatusEnum;
import com.opages.transaction.message.enums.PublicEnum;
import com.opages.transaction.message.model.TransactionMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;


/**
 * 消息定时器接口实现
 */
@Component("messageScheduled")
public class MessageScheduledImpl implements MessageScheduled, InitializingBean {

	private static final Log log = LogFactory.getLog(MessageScheduledImpl.class);
	ThreadPoolTaskExecutor threadPoolTaskExecutor;
	@Autowired
	private TransactionMessageApi transactionMessageApi;
	@Autowired
	private MessageDeal messageDeal;
	@Autowired
	private MessageProperties messageProperties;

	/**
	 * 处理状态为“待确认”但已超时的消息.
	 */
	public void handleWaitingConfirmTimeOutMessages() {
		try {
			int numPerPage = 2000; // 每页条数
			int maxHandlePageCount = 3; // 一次最多处理页数
			// 获取配置的开始处理的时间
			String dateStr = getCreateTimeBefore();
			TransactionMessage transactionMessage = new TransactionMessage();
			transactionMessage.setCreateTimeBefore(dateStr);
			transactionMessage.setStatus(MessageStatusEnum.WAITING_CONFIRM.value());
//			transactionMessage.setOrderBy(" insert_datetime ASC ");
			Map<String, TransactionMessage> messageMap = getMessageMap(numPerPage, maxHandlePageCount, transactionMessage);
			messageDeal.handleWaitingConfirmTimeOutMessages(messageMap);
		} catch (Exception e) {
			e.printStackTrace();
			log.error("处理[waiting_confirm]状态的消息异常" + e);
		}
	}


	/**
	 * 处理状态为“发送中”但超时没有被成功消费确认的消息
	 */
	public void handleSendingTimeOutMessage() {
		try {
			int numPerPage = 2000; // 每页条数
			int maxHandlePageCount = 3; // 一次最多处理页数
			// 获取配置的开始处理的时间
			String dateStr = getCreateTimeBefore();
			TransactionMessage transactionMessage = new TransactionMessage();
			transactionMessage.setCreateTimeBefore(dateStr);
			transactionMessage.setStatus(MessageStatusEnum.SENDING.value());
			transactionMessage.setAreadlyDead(PublicEnum.NO.value());
//			transactionMessage.setOrderBy(" insert_datetime ASC ");
			Map<String, TransactionMessage> messageMap = getMessageMap(numPerPage, maxHandlePageCount, transactionMessage);

			messageDeal.handleSendingTimeOutMessage(messageMap);
		} catch (Exception e) {
			e.printStackTrace();
			log.error("处理发送中的消息异常" + e);
		}
	}

	/**
	 * 根据分页参数及查询条件批量获取消息数据.
	 * @param numPerPage 每页记录数.
	 * @param maxHandlePageCount 最多获取页数.
	 * @param transactionMessage 查询参数.
	 * @return
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	private Map<String, TransactionMessage> getMessageMap(int numPerPage, int maxHandlePageCount, TransactionMessage transactionMessage){
		int pageNum = 1; // 当前页
		Map<String, TransactionMessage> messageMap = new HashMap<String, TransactionMessage>(); // 转换成map
		List<TransactionMessage> recordList = new ArrayList<TransactionMessage>(); // 每次拿到的结果集
//		int pageCount = 1; // 总页数
		log.info("==>pageNum:" + pageNum + ", numPerPage:" + numPerPage);
		recordList = transactionMessageApi.findList(pageNum,numPerPage,transactionMessage);
		if (recordList == null || recordList.isEmpty()) {
			log.info("==>recordList is empty");
			return messageMap;
		}
		log.info("==>now page size:" + recordList.size());

		for (TransactionMessage message : recordList) {
			messageMap.put(message.getUuid(), message);
		}
//		pageCount = page.getTotalPage(); // 总页数(可以通过这个值的判断来控制最多取多少页)
//		log.info("==>pageCount:" + pageCount);
//		if (pageCount > maxHandlePageCount){
//			pageCount = maxHandlePageCount;
//			log.info("==>set pageCount:" + pageCount);
//		}
		for (pageNum = 2; recordList!= null && recordList.size()>0; pageNum++) {
			log.info("==>pageNum:" + pageNum + ", numPerPage:" + numPerPage);
			recordList = transactionMessageApi.findList(pageNum,numPerPage,transactionMessage);
			if (recordList == null || recordList.isEmpty()) {
				break;
			}
			log.info("==>now page size:" + recordList.size());

			for (TransactionMessage message : recordList) {
				messageMap.put(message.getUuid(), message);
			}
		}
		recordList = null;
		return messageMap;
	}

	/**
	 * 获取配置的开始处理的时间
	 * @return
	 */
	private String getCreateTimeBefore() {
		String duration = messageProperties.getDuration();
		long currentTimeInMillis = Calendar.getInstance().getTimeInMillis();
		Date date = new Date(currentTimeInMillis - Integer.valueOf(duration) * 1000);
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		String dateStr = sdf.format(date);
		return dateStr;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		try {
			threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
			threadPoolTaskExecutor.setCorePoolSize(5);
			threadPoolTaskExecutor.setKeepAliveSeconds(3000);
			threadPoolTaskExecutor.setMaxPoolSize(10);
			threadPoolTaskExecutor.setQueueCapacity(100);
			threadPoolTaskExecutor.execute(new Runnable() {
				@Override
				public void run() {
					while (true) {
						log.info("执行(处理[待确认]状态的消息)任务开始");
						handleWaitingConfirmTimeOutMessages();
						log.info("执行(处理[待确认]状态的消息)任务结束");

						try {
							log.info("[待确认]睡眠60秒");
							Thread.sleep(60000);
						} catch (InterruptedException e) {
						}
					}
				}
			});
			threadPoolTaskExecutor.execute(new Runnable() {
				@Override
				public void run() {
					while (true) {
						log.info("执行(处理[发送中]的消息)任务开始");
						handleSendingTimeOutMessage();
						log.info("执行(处理[发送中]的消息)任务结束");
						try {
							log.info("[发送中]睡眠60秒");
							Thread.sleep(60000);
						} catch (InterruptedException e) {
						}
					}
				}
			});
		}catch (Exception e) {
			log.error("===>DubboReference context start error:", e);
		}
	}
}
